package com.atguigu.tingshu.album.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.RandomUtil;
import com.atguigu.tingshu.album.mapper.AlbumAttributeValueMapper;
import com.atguigu.tingshu.album.mapper.AlbumInfoMapper;
import com.atguigu.tingshu.album.mapper.AlbumStatMapper;
import com.atguigu.tingshu.album.mapper.TrackInfoMapper;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.common.cache.GuiGuCache;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.result.ResultCodeEnum;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.model.album.*;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import com.atguigu.tingshu.vo.album.AlbumAttributeValueVo;
import com.atguigu.tingshu.vo.album.AlbumInfoVo;
import com.atguigu.tingshu.vo.album.AlbumListVo;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Service
@SuppressWarnings({"all"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

    @Autowired
    private AlbumInfoMapper albumInfoMapper;

    @Autowired
    private AlbumAttributeValueMapper albumAttributeValueMapper;

    @Autowired
    private AlbumStatMapper albumStatMapper;

    @Autowired
    private TrackInfoMapper trackInfoMapper;

    @Autowired
    private KafkaService kafkaService;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 根据一级分类Id获取分类属性以及属性值（标签名，标签值）列表
     *
     * @param category1Id 一级分类ID
     * @return
     */
    @Override
    public List<BaseAttribute> getAttributeListByCategory1Id(Long category1Id) {
        //1.方式一：分别查询，先根据1级分类ID查询标签列表，循环标签列表分别查询每个标签包含标签值
        //2.方式二：编写动态SQL查询标签及值列表
        List<BaseAttribute> list = albumInfoMapper.getAttributeListByCategory1Id(category1Id);
        return list;
    }

    /**
     * 内容创作者或运营人员-保存专辑
     * 1.将提交专辑信息封装为专辑对象，向album_info表中新增一条记录
     * 2.将提交专辑标签封装为专辑标签关系集合，向album_attribute_value 表中新增若干条记录
     * 3.固定每个专辑产生四条初始化统计信息，向album_stat 表中新4增条记录
     *
     * @param albumInfoVo
     * @return
     */
    @Override
    @Transactional(rollbackFor = Exception.class)  //默认事务管理捕获运行时异常才会进行事务回滚
    public void saveAlbumInfo(AlbumInfoVo albumInfoVo, Long userId) {
        //1.将提交专辑信息封装为专辑对象，向album_info表中新增一条记录
        //1.1 拷贝VO实体类属性值到PO对象
        AlbumInfo albumInfo = BeanUtil.copyProperties(albumInfoVo, AlbumInfo.class);
        //1.2 为PO实体类属性赋值
        albumInfo.setUserId(userId);
        //包含声音数量
        albumInfo.setIncludeTrackCount(0);
        //VIP免费、付费类型专辑免费试听：5集
        albumInfo.setTracksForFree(5);
        //状态 0301-审核通过 0302-审核不通过 TODO 实际：专辑内容（文本、图片、声音）都应该经过审核
        albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
        //1.3 保存专辑
        albumInfoMapper.insert(albumInfo);
        Long albumId = albumInfo.getId();

        //2.将提交专辑标签封装为专辑标签关系集合，向album_attribute_value 表中新增若干条记录
        List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
        if (CollectionUtil.isNotEmpty(albumAttributeValueVoList)) {
            //将专辑标签VO集合转为PO集合
            albumAttributeValueVoList.stream().forEach(albumAttributeValueVo -> {
                //将AlbumAttributeValueVo 转为 AlbumAttributeValue
                AlbumAttributeValue albumAttributeValue = BeanUtil.copyProperties(albumAttributeValueVo, AlbumAttributeValue.class);
                // 标签关系关联标签ID
                albumAttributeValue.setAlbumId(albumId);
                //方式二：在遍历过程中单条新增
                albumAttributeValueMapper.insert(albumAttributeValue);
            });

            //TODO 方式一：调用专辑标签关系业务层AlbumAttributeValueService提供批量保存方法（效率高）
            //List<AlbumAttributeValue> albumAttributeValues = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
            //    //将AlbumAttributeValueVo 转为 AlbumAttributeValue
            //    AlbumAttributeValue albumAttributeValue = BeanUtil.copyProperties(albumAttributeValueVo, AlbumAttributeValue.class);
            //    // 标签关系关联标签ID
            //    albumAttributeValue.setAlbumId(albumId);
            //    return albumAttributeValue;
            //}).collect(Collectors.toList());
            //批量新增 TODO 作业
        }
        //3.固定每个专辑产生四条初始化统计信息，向album_stat 表中新4增条记录
        this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_PLAY);
        this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_SUBSCRIBE);
        this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_BUY);
        this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_COMMENT);

        //TODO 对专辑中内容（文本，图片）需要内容审核
        //4.判断如果专辑公开标识为1 发送Kafka消息完成专辑上架
        if ("1".equals(albumInfo.getIsOpen())) {
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER, albumId.toString());
        }
    }


    /**
     * 新增专辑统计信息
     *
     * @param albumId
     * @param statType
     */
    @Override
    public void saveAlbumStat(Long albumId, String statType) {
        AlbumStat albumStat = new AlbumStat();
        albumStat.setAlbumId(albumId);
        albumStat.setStatType(statType);
        albumStat.setStatNum(0);
        albumStatMapper.insert(albumStat);
    }


    /**
     * 查询当前创作者专辑分页列表
     *
     * @param pageInfo MP分页对象
     * @param query    查询条件（用户ID，关键字、状态）
     * @return
     */
    @Override
    public Page<AlbumListVo> getUserAlbumPage(Page<AlbumListVo> pageInfo, AlbumInfoQuery query) {
        //调用持久层动态SQL实现分页条件查询
        return albumInfoMapper.getUserAlbumPage(pageInfo, query);
    }

    /**
     * 根据专辑ID删除专辑信息
     *
     * @param id 专辑ID
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void removeAlbumInfo(Long id) {
        //1.先根据专辑ID查询专辑下声音数量 如果该专辑下包含声音，则不允许删除
        LambdaQueryWrapper<TrackInfo> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(TrackInfo::getAlbumId, id);
        Long count = trackInfoMapper.selectCount(queryWrapper);
        if (count > 0) {
            throw new GuiguException(ResultCodeEnum.NODE_ERROR);
        }
        //2.删除专辑记录
        albumInfoMapper.deleteById(id);

        //3.根据专辑ID删除专辑标签记录
        LambdaQueryWrapper<AlbumAttributeValue> albumAttributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>();
        albumAttributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
        albumAttributeValueMapper.delete(albumAttributeValueLambdaQueryWrapper);

        //4.根据专辑ID删除专辑统计记录
        LambdaQueryWrapper<AlbumStat> albumStatLambdaQueryWrapper = new LambdaQueryWrapper<>();
        albumStatLambdaQueryWrapper.eq(AlbumStat::getAlbumId, id);
        albumStatMapper.delete(albumStatLambdaQueryWrapper);

        //4.发送Kafka消息完成专辑下架
        kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, id.toString());
    }


    /***
     * 根据专辑ID查询专辑信息包含专辑标签列表
     * 使用自定义分布式锁：解决缓存击穿
     * @param id 专辑ID
     * @return
     */
   /* @Override
    public AlbumInfo getAlbumInfo(Long id) {
        try {
            //1.优先从缓存中获取数据
            //1.1 构建业务数据key 形式：前缀+主键ID
            String dataKey = RedisConstant.ALBUM_INFO_PREFIX + id;
            //1.2 从Redis中查询业务数据
            AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
            //1.3 命中缓存则直接返回即可，未命中则执行获取分布式锁逻辑
            if (albumInfo != null) {
                return albumInfo;
            }

            //2.尝试获取分布式锁
            //2.1 构建锁Key 形式：前缀+主键ID+后缀
            String lockKey = RedisConstant.ALBUM_INFO_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
            //2.2 构建锁Val UUID
            String lockVal = IdUtil.fastSimpleUUID();
            //2.3 采用set EX NX 实现分布式锁获取
            Boolean flag = redisTemplate.opsForValue().setIfAbsent(lockKey, lockVal, RedisConstant.ALBUM_LOCK_EXPIRE_PX2, TimeUnit.SECONDS);
            if (flag) {
                try {
                    //2.3 如果获取锁成功则执行业务
                    //3.执行业务（查询数据库、业务数据放入缓存）
                    albumInfo = this.getAlbumInfoFromDB(id);
                    //3.1 查询数据库为空：将空结果放入Redis 暂存10分钟
                    //3.2 查询数据库有值：将业务数据放入Redis 存1小时
                    int randomTTL = RandomUtil.randomInt(200, 1000);
                    long ttl = albumInfo == null ? RedisConstant.ALBUM_TEMPORARY_TIMEOUT : RedisConstant.ALBUM_TIMEOUT + randomTTL;
                    redisTemplate.opsForValue().set(dataKey, albumInfo, ttl, TimeUnit.SECONDS);
                    return albumInfo;
                } finally {
                    //4.释放分布式锁
                    //4.1 提供释放锁lua脚本
                    String text = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
                            "then\n" +
                            "    return redis.call(\"del\",KEYS[1])\n" +
                            "else\n" +
                            "    return 0\n" +
                            "end";
                    DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
                    redisScript.setScriptText(text);
                    redisScript.setResultType(Long.class);
                    //4.2 执行脚本
                    redisTemplate.execute(redisScript, Arrays.asList(lockKey), lockVal);
                }
            } else {
                //2.4 如果获取锁失败：自旋
                Thread.sleep(200);
                return this.getAlbumInfo(id);
            }
        } catch (Exception e) {
            //兜底处理：如果Redis锁服务不可用（Redis宕机了） ，直接查询数据库
            log.error("[专辑服务]获取专辑信息：{}", e);
            //throw new RuntimeException(e);
            return this.getAlbumInfoFromDB(id);
        }
    }*/


    /***
     * 根据专辑ID查询专辑信息包含专辑标签列表
     * 使用Redisson提供分布式锁：解决缓存击穿
     * @param id 专辑ID
     * @return
     */
    @Override
    public AlbumInfo getAlbumInfo(Long id) {
        try {
            //1.优先从缓存中获取数据
            //1.1 构建业务数据key 形式：前缀+主键ID
            String dataKey = RedisConstant.ALBUM_INFO_PREFIX + id;
            //1.2 从Redis中查询业务数据
            AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
            //1.3 命中缓存则直接返回即可，未命中则执行获取分布式锁逻辑
            if (albumInfo != null) {
                return albumInfo;
            }
            //2.获取分布式锁
            //2.1 声明锁名称 锁粒度设置小=为不同专辑设置不同锁
            String lockKey = RedisConstant.ALBUM_INFO_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
            //2.2 基于RedissonClient创建锁对象 入参锁名称就是存入Redis中锁Hash的key
            RLock lock = redissonClient.getLock(lockKey);
            //2.3 调用获取分布式锁方法-如果需要看门狗lock() 入参中leaseTime设置-1
            lock.lock();

            //3.执行业务
            try {
                //3.1 再次查询一次缓存：如果有大量线程已经调用lock方法，这些调用了lock方法线程最终都会获取锁成功，避免获取锁成功线程重复执行查询业务
                albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
                if (albumInfo != null) {
                    return albumInfo;
                }
                //3.2 执行查询数据库业务数据
                albumInfo = this.getAlbumInfoFromDB(id);

                //3.3 将查询结果放入缓存
                int randomTTL = RandomUtil.randomInt(200, 1000);
                long ttl = albumInfo == null ? RedisConstant.ALBUM_TEMPORARY_TIMEOUT : RedisConstant.ALBUM_TIMEOUT + randomTTL;
                redisTemplate.opsForValue().set(dataKey, albumInfo, ttl, TimeUnit.SECONDS);
                return albumInfo;
            } finally {
                //4.释放分布式锁
                lock.unlock();
            }
        } catch (Exception e) {
            //兜底处理：如果Redis锁服务不可用（Redis宕机了） ，直接查询数据库
            log.error("[专辑服务]获取专辑信息：{}", e);
            //throw new RuntimeException(e);
            return this.getAlbumInfoFromDB(id);
        }
    }


    /***
     * 根据专辑ID查询专辑信息包含专辑标签列表
     * @param id 专辑ID
     * @return
     */
    @Override
    @GuiGuCache(prefix = RedisConstant.ALBUM_INFO_PREFIX)
    public AlbumInfo getAlbumInfoFromDB(Long id) {
        //1.根据主键查询专辑信息
        AlbumInfo albumInfo = albumInfoMapper.selectById(id);

        //2.根据专辑ID查询专辑标签列表
        if (albumInfo != null) {
            LambdaQueryWrapper<AlbumAttributeValue> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
            List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueMapper.selectList(queryWrapper);
            albumInfo.setAlbumAttributeValueVoList(albumAttributeValueList);
        }
        return albumInfo;
    }


    /**
     * 修改专辑信息
     *
     * @param id          专辑ID
     * @param albumInfoVo 变更后专辑信息
     * @return
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void updateAlbumInfo(Long id, AlbumInfoVo albumInfoVo) {
        //1.修改专辑信息
        AlbumInfo albumInfo = BeanUtil.copyProperties(albumInfoVo, AlbumInfo.class);
        albumInfo.setId(id);
        albumInfoMapper.updateById(albumInfo);

        //2.修改专辑标签信息
        //2.1 先根据专辑ID将旧的专辑标签记录删除
        LambdaQueryWrapper<AlbumAttributeValue> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
        albumAttributeValueMapper.delete(queryWrapper);

        //2.2 在新增专辑标签记录
        List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
        if (CollectionUtil.isNotEmpty(albumAttributeValueVoList)) {
            albumAttributeValueVoList.stream().forEach(albumAttributeValueVo -> {
                AlbumAttributeValue albumAttributeValue = BeanUtil.copyProperties(albumAttributeValueVo, AlbumAttributeValue.class);
                //关联专辑ID
                albumAttributeValue.setAlbumId(id);
                //执行保存专辑标签
                albumAttributeValueMapper.insert(albumAttributeValue);
            });
        }

        //判断是否公开标识如果1 发送上架消息 如果flase 发送下架消息
        if ("1".equals(albumInfo.getIsOpen())) {
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER, id.toString());
        } else {
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, id.toString());
        }

    }

    /**
     * 查询用户发布专辑列表,用于下拉列表展示
     *
     * @param userId
     * @return
     */
    @Override
    public List<AlbumInfo> getUserAllAlbumList(Long userId) {
        //1.构建查询条件
        LambdaQueryWrapper<AlbumInfo> queryWrapper = new LambdaQueryWrapper<>();
        // 用户ID
        queryWrapper.eq(AlbumInfo::getUserId, userId);
        // 限制查询字段
        queryWrapper.select(AlbumInfo::getId, AlbumInfo::getAlbumTitle);
        // 按照ID降序
        queryWrapper.orderByDesc(AlbumInfo::getId);
        // SQL 最后拼接 limit 200
        queryWrapper.last("limit 200");

        //2.执行查询专辑表
        return albumInfoMapper.selectList(queryWrapper);
    }

    /**
     * 根据专辑ID查询专辑统计信息
     *
     * @param albumId 专辑ID
     * @return
     */
    @Override
    @GuiGuCache(prefix = "albumStatVo:")
    public AlbumStatVo getAlbumStatVo(Long albumId) {
        return albumInfoMapper.getAlbumStatVo(albumId);
    }
}
